package com.atguigu.day03;

import com.atguigu.utils.ClickEvent;
import com.atguigu.utils.ClickSource;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

// create database userbehavior;
// use userbehavior;
// create table clicks (username varchar(100), url varchar(100));
public class Example4 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .addSink(new MyJDBC());

        env.execute();
    }

    // RichSinkFunction<输入的泛型>
    public static class MyJDBC extends RichSinkFunction<ClickEvent> {
        private Connection connection;
        private PreparedStatement insertStmt;
        private PreparedStatement updateStmt;

        @Override
        public void open(Configuration parameters) throws Exception {
            // 获取一个数据库的连接
            connection = DriverManager.getConnection(
                    "jdbc:mysql://localhost:3306/userbehavior?useSSL=false",
                    "root",
                    "root"
            );

            insertStmt = connection.prepareStatement(
                    "insert into clicks (username, url) values (?,?)"
            );

            updateStmt = connection.prepareStatement(
                    "update clicks set url = ? where username = ?"
            );
        }

        // 每来一条数据触发一次执行
        @Override
        public void invoke(ClickEvent input, Context context) throws Exception {
            // 幂等性写入mysql，每个username只有一条数据
            // 如果存在username的数据，则更新
            // 如果不存在username的数据，则插入

            // 先试图更新
            updateStmt.setString(1, input.url);
            updateStmt.setString(2, input.username);
            updateStmt.execute();

            // 如果更新的行数是0,说明input.username对应的数据在表中不存在
            // 插入数据
            if (updateStmt.getUpdateCount() == 0) {
                insertStmt.setString(1, input.username);
                insertStmt.setString(2, input.url);
                insertStmt.execute();
            }
        }

        @Override
        public void close() throws Exception {
            insertStmt.close();
            updateStmt.close();
            connection.close();
        }
    }
}
